/**
 * 1、基本模式
 * 2、工作模式
 */
package com.example.rabbitMQ;

import com.rabbitmq.client.*;
import java.io.IOException;

public class BasicDemo {
    static Connection connection = null;
    static Channel channel = null;

    //建立与RabbitMQ Server的连接
    public static Connection getConnection() throws Exception {
        //定义连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        //设置服务地址
        factory.setHost("localhost");
        //端口
        factory.setPort(5672);
        //设置账号信息，用户名、密码、vhost
        factory.setUsername("guest");
        factory.setPassword("guest");
        // 通过工程获取连接
        Connection connection = factory.newConnection();
        return connection;
    }

    //-----------简单模式（基本消费者），点对点发送和接收消息-----------
    private static final String QUEUE_NAME1 = "basic_queue";

    public static void basicProduce() {
        try {
            //1、获取连接,消息发送端与mq服务创建连接
             connection = getConnection();
            //2、建立信道
             channel = connection.createChannel();
            /*3、声明队列,队列需要提前声明，如果未声明就使用队列，参数明细
             * 1）queue 队列名称
             * 2）durable 是否持久化，如果持久化，mq重启后队列还在
             * 3）exclusive 是否独占连接，队列只允许在该连接中访问，如果connection连接关闭队列则自动删除,如果将此参数设置true可用于临时队列的创建
             * 4）autoDelete 自动删除，队列不再使用时是否自动删除此队列，如果将此参数和exclusive参数设置为true就可以实现临时队列（队列不用了就自动删除）
             * 5）arguments 参数，可以设置一个队列的扩展参数，比如：可设置存活时间
             */
            channel.queueDeclare(QUEUE_NAME1, false, false, false, null);

            //4、创建消息
            String message = "Hello 基本消费者模式!";

            /*5、发布消息，参数明细：
             * 1）exchange，交换机，如果不指定将使用mq的默认交换机（设置为""）
             * 2）routingKey，路由key，交换机根据路由key来将消息转发到指定的队列，如果使用默认交换机，routingKey设置为队列的名称
             * 3）props，消息的属性
             * 4）body，消息内容
             */
            channel.basicPublish("", QUEUE_NAME1, null, message.getBytes());

            System.out.println("[BasicQueue点对点生产者]sent：" + message);
            //6、关闭信道
            channel.close();
            //7、关闭连接
            connection.close();

        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    //消费时自动返回确认
    public static void basicConsume() {
        try {
        //1、获取连接,消息消费者与mq服务建立连接
        connection =  getConnection();
        //2、建立信道
        channel = connection.createChannel();
        //3、声明队列,队列需要提前声明，如果未声明就使用队列
        channel.queueDeclare(QUEUE_NAME1, false, false, false, null);

        //4、定义队列的消费者
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            // 6、获取消息，并且处理，这个方法类似事件监听，如果有消息的时候，会被自动调用
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
                                       byte[] body) throws IOException {
                // body 即消息体
                String msg = new String(body);
                System.out.println("[BasicQueue点对点消费者]received：" + msg);
            }
        };

        /*5、监听队列,自动ACK(消息一旦被接收，消费者自动发送ACK,第二个参数是true，适用于消息不太重要，丢失也没有影响的场合）,参数明细：
         * 1)queue 队列名称
         * 2)autoAck  自动回复，当消费者接收到消息后要告诉mq消息已接收，如果将此参数设置为tru表示会自动回复mq，如果设置为false要通过编程实现回复
         * 3)callback，消费方法，当消费者接收到消息要执行的方法
         */
        channel.basicConsume(QUEUE_NAME1, true, consumer);

    }   catch (Exception e) {
        e.printStackTrace();
    }
}
    //-----------工作模式（竞争消费者），点对多点消息发送和接收------------
    //单队列多消费者，通常队列会以轮询的方式将消息分别交给各个消费者去处理
    //但是如果消费者群里面的消费消息的能力不均匀的话，可以将basicQos设置为1，这样每个消费者在ack之后。队列才会分配给它下一条消息
    private static final String QUEUE_NAME2 = "work_queue";
    public static void workqueueProduce() {
        try{
            //消息发送端与mq服务创建连接
            connection =  getConnection();
            //建立通道
            channel = connection.createChannel();
            channel.queueDeclare(QUEUE_NAME2, false, false, false, null);
            //循环发布消息
            for (int i = 1; i <= 10; i++) {
                // 消息内容
                String message = "Hello 竞争消费者模式----" + i;
                channel.basicPublish("", QUEUE_NAME2, null, message.getBytes());
                System.out.println("[WorkQueue生产者]sent：" + message);
                Thread.sleep(500);
            }

            channel.close();
            connection.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    //消费时手动返回确认
    public static void workqueueConsume1() {
        try {
            connection =  getConnection();
            channel = connection.createChannel();
//            channel.queueDeclare(QUEUE_NAME2, false, false, false, null); //可选，此处声明队列的目的是避免报错

            //开启能者多劳模式：同一时刻服务器只会发送一条消息给消费者;消费者在接收到队列里的消息但没有返回确认结果之前,队列不会将新的消息分发给该消费者
            //队列中没有被消费的消息不会被删除，还是存在于队列中;只有在channel.basicQos被使用的时候channel.basicAck(delivery.getEnvelope().getDeliveryTag(),false)才起到作用
            channel.basicQos(1);
            DefaultConsumer consumer = new DefaultConsumer(channel) {
                // 获取消息，并且处理，这个方法类似事件监听，如果有消息的时候，会被自动调用
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
                                           byte[] body) throws IOException {
                    // body 即消息体
                    String msg = new String(body);
                    System.out.println("[WorkQueue消费者---1]received：" + msg);
                    try {
                        Thread.sleep(100);//模拟消费耗时
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                     //返回确认状态,手动确认,false代表不开启多次确认
                     channel.basicAck(envelope.getDeliveryTag(), false);

                     //拒绝消费当前消息,可以将消息数据重新放回队列，以供再次消费（true）或者直接丢弃（false）
                     //channel.basicReject(envelope.getDeliveryTag(), true);
                     //第一个参数是当前消息数据的唯一id
                     //第二个参数是指是否针对多条消息；如果是true，也就是说一次性针对当前通道的消息的tagID小于当前这条消息的，都拒绝确认
                     //第三个参数是指是否重新入列，也就是指不确认的消息是否重新丢回到队列里面去
                     //channel.basicNack(envelope.getDeliveryTag(), false, true);
                }
            };

            /*监听队列,手动ACK(消息一旦被接收，消费者不会发送ACK,而需调用channel.basicAck发送确认，第二个参数是false，适用于消息非常重要，不容丢失的场合）,参数明细：
             * 1)queue 队列名称
             * 2)autoAck  自动回复，当消费者接收到消息后要告诉mq消息已接收，如果将此参数设置为tru表示会自动回复mq，如果设置为false要通过编程实现回复
             * 3)callback，消费方法，当消费者接收到消息要执行的方法
             */
            channel.basicConsume(QUEUE_NAME2, false, consumer);

        }   catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static void workqueueConsume2() {
        try {
            //消息消费者与mq服务建立连接
            connection =  getConnection();
            //建立通道
            channel = connection.createChannel();
//          channel.queueDeclare(QUEUE_NAME2, false, false, false, null);    //可选，此处声明队列的目的是避免报错
            channel.basicQos(1);
            DefaultConsumer consumer = new DefaultConsumer(channel) {
                // 获取消息，并且处理，这个方法类似事件监听，如果有消息的时候，会被自动调用
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
                                           byte[] body) throws IOException {
                    // body 即消息体
                    String msg = new String(body);
                    System.out.println("[WorkQueue消费者---2]received：" + msg);
                    try {
                        Thread.sleep(300);//模拟消费耗时
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    //进行手动确认消息 false代表不开启多次确认
                    channel.basicAck(envelope.getDeliveryTag(), false);
                }
            };
            channel.basicConsume(QUEUE_NAME2, false, consumer);
        }   catch (Exception e) {
            e.printStackTrace();
        }
    }
}
